import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2019/9/2 10:37
  * @ Param:  ${PARAM}
  * @ Description:
  */
object test {
  /**
    * 新增地址
    * @param args
    */
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.set("spark.debug.maxToStringFields", "500")
    conf.setMaster("local[*]")
    conf.set("es.nodes", "192.168.1.157")
    conf.set("es.port", "9200")
    conf.set("cluster.name", "O2OElastic")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.set("spark.sql.caseSensitive", "true")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    sc.setLogLevel("WARN")
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")


//    sqlContext.read.json(s"s3a://o2o-dataproces-group/chen_lixiu/2020/4/miya/good/").registerTempTable("t1")
//    sqlContext.read.orc(s"s3a://o2o-dataproces-group/zhang_feng/dws-data/esdata/2020/wangyiyanxuan/4/").registerTempTable("t1")
//    sqlContext.read.orc(s"s3a://o2o-dataproces-group/zhang_feng/dws-data/esdata/2020/wangyiyanxuan/4/").registerTempTable("t1")
//    sqlContext.read.orc(s"s3a://o2o-dataproces-group/li_yinchao/lyc/platform/jumei/2020/5/jumei_2020_5/").registerTempTable("t2")
//    sqlContext.read.orc(s"s3a://o2o-dataproces-group/li_yinchao/lyc/platform/jumei/2020/4/brand_join_text_2/").registerTempTable("t1")
//    sqlContext.read.orc(s"s3a://o2o-dataproces-group/zsc/2020/5/miya/good/").registerTempTable("t2")
   /* sqlContext.sql(
      """
        |select *,CAST(sum(sellCount) as bigint) sell,CAST(sum(salesAmount) as dei sale
        |from t2
      """.stripMargin).show()
    sqlContext.sql(
      """
        |select *,sum(sellCount) sell,sum(salesAmount) sale
        |from t1
      """.stripMargin).show()*/

/*    sqlContext.sql(
      """
        |select count(1)
        |from t2
      """.stripMargin).show()

    sqlContext.sql(
      """
        |select count(1)
        |from t1
      """.stripMargin).show()*/
    //|   20128|1053853|85413439.39|

//          val brandTableResult = s"s3a://o2o-dataproces-group/li_yinchao/Table/2020/${month+1}/${platformName}"

//          val brandNewAddTable = s"s3a://o2o-dataproces-group/li_yinchao/Table/newAddBrand/${year}/${month}/${platformName}"
//    for (platform <- Array("dazhongdp", "elm", "jd", "meituan", "meituan_tg", "pinduoduo", "suning", "taobao", "tmall")) {
//    for (platform <- Array("meituan", "elm", "meituan_tg", "pinduoduo", "suning", "taobao", "tmall")) {
////    for (platform <- Array("dazhongdp", "jd", "meituan_tg", "pinduoduo", "suning", "taobao", "tmall")) {
//
//      val value = sqlContext.read.orc(s"s3a://dws-data/g_shop/618/20190618_2020_yoy/${platform}")
//        .drop("timeStamp")
//        .withColumn("timeStamp", lit("1561824000")).write.orc(s"s3a://dws-data/g_shop/618/20190618_2020_yoy/${platform}_v1")
//
//
//    }






     /* sqlContext.sql(
        """
          |select
          |a.*
          |from
          |t1 a
        """.stripMargin
      ).repartition(1).write.orc()*/
    }

//    sqlContext.sql(
//      """
//        |
//        |select
//        |count(1)
//        |from t1
//        |
//      """.stripMargin).show()
//    sqlContext.sql(
//      """
//        |
//        |select
//        |count(1),
//        |sum(sellCount)
//        |from t2
//        |
//      """.stripMargin).show()




    /*val data=sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/taobao/${years}/${months}/zhibo_finally_v1")
    import org.apache.spark.sql.functions._
    data.agg(count("*"),sum("sellCount"),sum("salesAmount")).show(false)*/

    /*val data1 = sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/zhibo_zf/zhubo_v1/12")
    val data2 = sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/zhibo_zf/zhubo/12")

    data1.printSchema()
    data2.printSchema()

    data1.where("platformId in ('-1')").show(false)*/
    //println(data1.count())
    //println(data2.count())

    //data1.show(false)

    //val tb_all_data = sqlContext.read.json("s3a://o2o-dataproces-group/panzonghao/taobao/all_userid/*/*").dropDuplicates()
    //tb_all_data.createOrReplaceTempView("tb_all_data")

    /*for (months <- 1 to 12){
      val value: RDD[String] = sc.esJsonRDD(s"2019_tmall_anchor/tmall_anchor_2019_${months}").values
      sqlContext.read.json(value).createOrReplaceTempView("tm_data")

      val value1: RDD[String] = sc.esJsonRDD(s"2019_taobao_anchor/taobao_anchor_2019_${months}").values
      sqlContext.read.json(value1).createOrReplaceTempView("tb_data")

      sqlContext.sql(
        s"""
           |select
           |t1.anchorId
           |from (select * from tm_data where is_selfLive='true' and is_selfShopAnchor='false') t1
           |left join
           |tb_data t2
           |on t1.anchorId=t2.anchorId
           |where t2.anchorId is null
         """.stripMargin).show(false)
    }
*/



    //println(xx.unionAll(xx1).count())

    /*for (months <- 1 to 1){
      val data = sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/zhibo_zf/zhubo/${months}/")
      data.printSchema()

      println("**********")
      println(data.count())
    }
*/



    //println(result.count())
   // println(dp_zhubo.count())
    //println(result.where("is_selfLive = 'true' and province!='0'").count())
    //println(dr_zhubo.count())
    //println(result.where("is_selfLive = 'false' or (is_selfLive = 'true' and province='0')").count())


    /*result.dropDuplicates().createOrReplaceTempView("result")
    dp_zhubo.union(dr_zhubo).dropDuplicates().createOrReplaceTempView("dp_zhubo")

    sqlContext.sql(
      s"""
         |select
         |t1.*
         |from result t1
         |left join
         |dp_zhubo t2
         |on t1.anchorId=t2.anchorId
         |where t2.anchorId is null
       """.stripMargin).show(false)*/
    /*for (months <- 1 to 1){
      val data = sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/zhibo_zf/zhubo/${months}/")
      println(data.dropDuplicates("anchorId").count())

    }*/



   /* data.printSchema()
    data.where("is_liveShop='true'").show(200,false)*/


    /*var b = "false"
    breakable {
      for (i <- 0 to 8){
        if (i==8){
          b = "true"
          break
        }
      }
    }
    print(b)*/
    /*val data = sqlContext.read.orc("s3a://o2o-dataproces-group/panzonghao/tmall/12/zhibo_zhubo")
    data.printSchema()
    println(data.count())*/

/*
    for (months <- 1 to 12){
      val data = sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/zhibo_zf/good/${months}")
      /*data.printSchema()
      data.show(false)*/
      import org.apache.spark.sql.functions._
      data.agg(count("*"),sum("sellCount"),sum("salesAmount")).show(false)
    }*/


   /* import org.elasticsearch.spark._

    val data = sc.esJsonRDD(s"${indexs}",
      """
        |{
        |  "query": {
        |    "bool": {
        |      "should": [
        |        {
        |          "term": {
        |            "thirdCategoryId": {
        |              "value": "100210502"
        |            }
        |          }
        |        },
        |        {
        |          "term": {
        |            "thirdCategoryId": {
        |              "value": "100210501"
        |            }
        |          }
        |        }
        |      ]
        |    }
        |  }
        |}
      """.stripMargin).values

    data.saveAsTextFile(s"s3a://o2o-dataproces-group/panzonghao/tmall/12/suannai_backs")

    niunai_chai(sc.esJsonRDD(es_sql()._5,es_sql()._3).values,sqlContext.sparkSession)
    println("*"*50)
    suannai_chai(sc.esJsonRDD(es_sql()._5,es_sql()._4).values,sqlContext.sparkSession)*/
    /*val data = sqlContext.read.json(s"s3a://o2o-sourcedata/obs-source-2019/12/tmall/tmall_mmm_app_1912_2/")

    data.show(false)*/
    //println(data.where("is_newProduct='true'").count())

    /*println(data.selectExpr("good_id").count())
    println(data.selectExpr("good_id").dropDuplicates().count())*/
      //.where("good_id='607929237063'").selectExpr("priceText","sellCount","salesAmount")//.show(false)
    /*val data = sqlContext.read.json(s"s3a://o2o-sourcedata/obs-source-2019/12/tmall/tmall_mmm_app_1912/")

    data.show(200,false)
*/
    //data.agg(count("*"),sum("sellCount"),sum("salesAmount")).show(false)
   //data.selectExpr("sum(sellCount)","sum(sellCount * priceText)","count(good_id)").show(111)

    //println(min_price("4","-1"))

    /*sqlContext.read.orc("s3a://o2o-dataproces-group/yang_songjian/product/taobao/2019/12/category/cate_1",
      "s3a://o2o-dataproces-group/yang_songjian/product/taobao/2019/12/category/cate_2",
      "s3a://o2o-dataproces-group/yang_songjian/product/taobao/2019/12/category/cate_3",
      "s3a://o2o-dataproces-group/yang_songjian/product/taobao/2019/12/category/cate_4",
      "s3a://o2o-dataproces-group/yang_songjian/product/taobao/2019/12/category/cate_5",
      "s3a://o2o-dataproces-group/yang_songjian/product/taobao/2019/12/category/cate_6"
    ).repartition(4).write.json("s3a://o2o-dataproces-group/panzonghao/hancongcong/cate")
*/


   /*val data = sqlContext.read.json(s"s3a://o2o-dataproces-group/panzonghao/2019/${month}/all_good_v1")

    data.printSchema()
    import org.apache.spark.sql.functions._
    data.agg(count("*"),sum("sellCount"),sum("salesAmount")).show(false)*/

    //println(tranTimeToString("1569772800"))
    //val taobao = sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/taobao/11/zhibo_zhubo_v5_test/")
    //val tmall = sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/tmall/11/zhibo_zhubo_v5_test/")

   /* taobao.createOrReplaceTempView("taobao")
    tmall.createOrReplaceTempView("tmall")
*/

   /* val data = sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/taobao/5/zhibo_finally_month5")

    println("6月份总条数为:"+data.count())

    for (months <- 1 to 11){
     /* val result = sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/tmall/${months}/tmall_live_detail_191101_before")
        .dropDuplicates("good_id")

      println(s"${months}月总数为:"+result.count())*/


    }*/


    /*taobao.toJSON.rdd.map(line =>{
      JSON.parseObject(line)
    }).saveToEs(s"2019_taobao_anchor/taobao_anchor_2019_11", Map("es.mapping.id" -> "anchorId"))

    tmall.toJSON.rdd.map(line =>{
      JSON.parseObject(line)
    }).saveToEs(s"2019_tmall_anchor/tmall_anchor_2019_11", Map("es.mapping.id" -> "anchorId"))*/



//////////////////////////////////////////////////////////////////////////////////////////////////

    /*for (month <- 1 to 10){
      val result = sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/taobao/${month}/zhibo_finally/")

      result.printSchema()
      import org.apache.spark.sql.functions._
      result.agg(count("*"),sum("sellCount"),sum("salesAmount")).show(false)
    }*/

    //val result = sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/taobao/11/zhibo_zhubo_v5/")

    /*println("***********"+result.where("cast(sellCount as double)=0").count())
    result.createOrReplaceTempView("result")
    //result.printSchema()
    //result.selectExpr("sellCount","salesAmount").show()
    result.where("anchorId='69226163'").show(false)
    println("*******"+result.count())
    sqlContext.sql(
      s"""
         |select
         |anchorId,
         |sellCount,
         |salesAmount
         |from result
         |order by salesAmount desc
       """.stripMargin).show(false)

    import org.apache.spark.sql.functions._
    result.agg(sum("salesAmount")).show(false)*/
    //入库
  /*  result.toJSON.rdd.map(line =>{
      JSON.parseObject(line)
    }).saveToEs(s"2019_taobao_anchor/taobao_anchor_2019_11", Map("es.mapping.id" -> "anchorId"))*/
///////////////////////////////////////////////////////////////////////////////////////////////////////////



    /*val result = sqlContext.read.orc("s3a://o2o-dataproces-group/panzonghao/taobao/11/zhibo_finally")
      result.createOrReplaceTempView("result")

    import org.apache.spark.sql.functions._
    result.agg(count("*"),sum("sellCount"),sum("salesAmount")).show(false)*/
    /*result.createOrReplaceTempView("result")
    println("******"+result.count())*/
    /*sqlContext.sql(
      s"""
         |select
         |*,
         |cast(liveId as String) as liveIds
         |from result
       """.stripMargin).drop("liveId").withColumnRenamed("liveIds","liveId")*/

    /*sqlContext.read.orc("s3a://o2o-dataproces-group/panzonghao/tmall/11/zhibo_live_good_v1")
    .withColumn("timeStamp",lit("1575043200"))
    .withColumn("platformId",lit("10"))
      .repartition(4).write.orc("s3a://o2o-dataproces-group/panzonghao/tmall/11/zhibo_live_good_v2")*/


   /* import org.elasticsearch.spark._

    val result = sqlContext.read.orc("s3a://o2o-dataproces-group/panzonghao/tmall/11/zhibo_live_good_v5")

    result.toJSON.rdd.map(line =>{
      JSON.parseObject(line)
    }).saveToEs(s"2019_tmall_good_live/tmall_good_live_2019_11", Map("es.mapping.id" -> "liveAndgood_id"))


    val result1 = sqlContext.read.orc("s3a://o2o-dataproces-group/panzonghao/taobao/11/zhibo_live_good_v5")

    result1.toJSON.rdd.map(line =>{
      JSON.parseObject(line)
    }).saveToEs(s"2019_taobao_good_live/taobao_good_live_2019_11", Map("es.mapping.id" -> "liveAndgood_id"))
*/


    //data.createOrReplaceTempView("data")

    /*sqlContext.sql(
      s"""
         |select
         |count(*)
         |from data
       """.stripMargin).show(false)
    data.printSchema()*/




    /*import org.apache.spark.sql.functions._
    data.agg(count("*")).show(false)*/

    //data.show()


    /*data.dropDuplicates("liveId").withColumnRenamed("fans","anchorFans")
      .repartition(4).write.orc("s3a://o2o-dataproces-group/panzonghao/tmall/11/zhibo_zhubo")*/




  }

//  def tranTimeToString(timestamp:String) :String={
//    val tm = timestamp+"000"
//    val fm = new SimpleDateFormat("yyyy-M")
//    val time = fm.format(new Date(tm.toLong))
//    time
//  }

//  def min_price(price1:String,price2:String):String={
//    scala.math.min(price1.toDouble,price2.toDouble).toString
//  }

//}
